package com.assistant.message;

import cn.hutool.crypto.digest.MD5;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.assistant.dao.domain.SysOperLog;
import com.assistant.dto.CanalMessage;
import com.assistant.es.SysOperLogDto;
import com.assistant.service.es.ElasticOperatorService;
import com.assistant.util.DateUtil;
import com.assistant.util.MD5Util;
import com.google.common.io.BaseEncoding;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.dubbo.common.utils.MD5Utils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.MDC;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

@Component
@Slf4j
public class CanalDbLogConsumer {
    @Value("${kafka.max.consumer.count:3}")
    private Integer maxConsumerCount;
    @Autowired
    private StringRedisTemplate redisTemplate;
    @Autowired
    private ElasticOperatorService elasticOperatorService;
    /**
     * 线程ID常量
     */
    public static final String THREAD_ID = "THREAD_ID";
	// 通过注解注入所在的消费组和要监听的Topic
    @KafkaListener(topics = "${kafka.consumer.canal-db-log.topic}",
            groupId = "${kafka.consumer.canal-db-log.group-id}",
            concurrency = "1",
            autoStartup = "${kafka.consumer.autostart}")
    public void topicConsumer(ConsumerRecord<String, String> record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) Object topic) {
        String value = record.value();
        String key  = MD5Util.getMD5(value);
        try {
            JSONObject jsonObject = JSONObject.parseObject(value);
            CanalMessage canalMessage = JSON.toJavaObject(jsonObject, CanalMessage.class);
            log.info("消费kafka-canal: 消息={}",JSON.toJSONString(canalMessage));
            String tableName = canalMessage.getTable();
            switch (canalMessage.getType()) {
                case "INSERT":
                case "UPDATE":
                    if ("sys_oper_log".equals(tableName)) {
//                        sysOperLogHandle(canalMessage.getData());
                    }
                    break;
                case "DELETE":
                    break;
                case "TRUNCATE":
                    break;
                default:
                    break;
            }
            ack.acknowledge();
        } catch (Exception e) {
            log.error("消费kafka消息失败-registerUserMessageLisener:{}", e);
            boolean skip = skipError(key, value);
            if (skip) {
                ack.acknowledge();
            } else {
                ack.nack(5000L);
            }
        }
    }
    private void sysOperLogHandle(List<JSONObject> data) {
        // 增加、修改处理
        for (JSONObject jsonObject : data) {
            SysOperLogDto sysOperLogDto = new SysOperLogDto();
            String oper_id = jsonObject.getString("oper_id");
            log.info("保存操作日志索引:{}", sysOperLogDto);
            elasticOperatorService.save(sysOperLogDto);
        }
    }

    /**
     * 是否跳过错误消费
     *
     * @param key
     * @param value
     * @return
     */
    private boolean skipError(String key, String value) {
        if (StringUtils.isEmpty(key)) {
            return true;
        }
        Long increment = redisTemplate.opsForValue().increment(key, 1l);
        redisTemplate.expire(key, 1l, TimeUnit.MINUTES);
        log.info("消费kafka消息次数increment:{}", increment);
        if (increment >= maxConsumerCount) {
            log.error("消费kafka消息三次,中断:key:{},value:{}", key, value);
            return true;
        }
        return false;
    }
}
